package org.gbif.occurrence.store.util;

import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import java.util.Map.Entry;
import java.util.regex.Pattern;

import org.apache.hadoop.hbase.client.Put;

/**
 * Helps read a row from a fielded test file using properties 
 * to look up the mapping.  This allows for name based lookup on a 
 * (e.g.) tab delimited file by providing the tab file descriptor 
 * 
 * properties should be like:
 * scientificName=1
 * latitude=2
 * 
 * @author tim
 */
public class ConfigurableRecordReader {
	protected Pattern tabPattern = Pattern.compile("\t");
	protected Pattern colonPattern = Pattern.compile(":");
	protected Properties p = new Properties();
	protected boolean ignoreNulls = false;
	
	public ConfigurableRecordReader(String file) throws IOException {
		p.load(ConfigurableRecordReader.class.getResourceAsStream(file));
	}
	
	public ConfigurableRecordReader(String file, boolean ignoreNulls, String separator) throws IOException {
		p.load(ConfigurableRecordReader.class.getResourceAsStream(file));
		this.ignoreNulls = ignoreNulls;
		tabPattern = Pattern.compile(separator);
	}
	
	public ConfigurableRecordReader(String file, boolean ignoreNulls) throws IOException {
		p.load(ConfigurableRecordReader.class.getResourceAsStream(file));
		this.ignoreNulls = ignoreNulls;
	}
	
	public String[] split(String line) {
		return tabPattern.split(line);
	}
	
	// take the key, find the index in the row of data to use 
	protected Integer getPartIndex(String key) {
		try {
			return Integer.parseInt(p.getProperty(key));
		} catch (NumberFormatException e) {
			return null;
		}
	}
	
	/**
	 * This will read all the values defined in the mapping found in the splits and
	 * will read into the output provided
	 * Anything not found is ignored.
	 * @return Number of properties mapped
	 */
	public int readAllInto(String splits[], Map<String, String> output) {
		int count = 0;
		for (Entry<Object, Object> propertyMapping: p.entrySet()) {
			try {
				int index = Integer.parseInt(propertyMapping.getValue().toString());
				output.put(propertyMapping.getKey().toString(), splits[index]);
				count++;
			} catch (Exception e) {
				// swallow bad mappings and no data
			}
		}
		return count;
	}
	
	/**
	 * This will read all the values defined in the mapping found in the splits and
	 * will add them to the row
	 * Anything not found is ignored.
	 * @return Number of properties mapped
	 */
	public int readAllInto(String splits[], Put row) {
		int count = 0;
		for (Entry<Object, Object> propertyMapping: p.entrySet()) {
			try {
				int index = Integer.parseInt(propertyMapping.getValue().toString());
				String[] familyColumn = colonPattern.split(propertyMapping.getKey().toString());
				row.add(familyColumn[0].getBytes(), familyColumn[1].getBytes(), splits[index].getBytes());
				count++;
			} catch (Exception e) {
				// swallow bad mappings and no data
			}
		}
		return count;
	}
	
	public String getString(String key, String[] splits) {
		Integer index = getPartIndex(key);
		try {
			if (index!=null) {
				if (ignoreNulls) {
					String value = splits[index];
					if ("\\N".equals(value)) {
						return null;
					} else {
						return value; 
					}
				} else {
					return splits[index];
				}
			}
		} catch (ArrayIndexOutOfBoundsException e) {
		}
		return null;
	}
	
	public Float getFloat(String key, String splits[]) {
		String valueAsString = getString(key, splits);
		if (valueAsString != null) {
			try {
				return Float.parseFloat(valueAsString);
			} catch (NumberFormatException e) {
			}
		}
		return null;
	}
	
	public Integer getInteger(String key, String splits[]) {
		String valueAsString = getString(key, splits);
		if (valueAsString != null) {
			try {
				return Integer.parseInt(valueAsString);
			} catch (NumberFormatException e) {
			}
		}
		return null;
	}
	
	public Long getLong(String key, String splits[]) {
		String valueAsString = getString(key, splits);
		if (valueAsString != null) {
			try {
				return Long.parseLong(valueAsString);
			} catch (NumberFormatException e) {
			}
		}
		return null;
	}
	
	public Double getDouble(String key, String splits[]) {
		String valueAsString = getString(key, splits);
		if (valueAsString != null) {
			try {
				return Double.parseDouble(valueAsString);
			} catch (NumberFormatException e) {
			}
		}
		return null;
	}
}
